﻿using System;
using System.Collections.Concurrent;
using System.Data.SQLite;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Protocol;
using Serilog;
using System.Collections.Generic;

public class PublishStrategyManager
{
    private readonly MqttClientWrapper _mqttClient;
    private readonly ConcurrentDictionary<string, string> _lastValues = new ConcurrentDictionary<string, string>();
    private readonly ConcurrentDictionary<string, (DateTime lastPublishTime, string value)> _lastPublishTimes = new ConcurrentDictionary<string, (DateTime, string)>();
    private readonly ConcurrentDictionary<string, List<string>> _accumulatedData = new ConcurrentDictionary<string, List<string>>();
    private readonly ConcurrentDictionary<string, (string value, int priority)> _priorityData = new ConcurrentDictionary<string, (string, int)>();
    private Timer _timer;
    private const int AccumulationThreshold = 10;
    // 添加批量发布缓存
    private readonly Channel<(string topic, string payload, MqttQualityOfServiceLevel qos, bool retain)> _publishChannel = 
        Channel.CreateBounded<(string, string, MqttQualityOfServiceLevel, bool)>(1000);
    private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(Environment.ProcessorCount * 2);
    public PublishStrategyManager(MqttClientWrapper mqttClient)
    {
        _mqttClient = mqttClient;
        // 启动后台发布任务
        Task.Run(ProcessPublishQueue);
    }
    // 后台处理发布队列
    private async Task ProcessPublishQueue()
    {
        while (await _publishChannel.Reader.WaitToReadAsync())
        {
            while (_publishChannel.Reader.TryRead(out var message))
            {
                await _semaphore.WaitAsync();
                try
                {
                    await _mqttClient.PublishAsync(message.topic, message.payload, message.qos, message.retain);
                }
                finally
                {
                    _semaphore.Release();
                }
            }
        }
    }
    // 优化后的数据变化发布策略
    public async ValueTask OnDataChanged(string tag, string value)
    {
        if (_lastValues.TryGetValue(tag, out var lastValue) && lastValue != value)
        {
            await _publishChannel.Writer.WriteAsync((tag, value, MqttQualityOfServiceLevel.AtLeastOnce, false));
            _lastValues[tag] = value;
        }
        else if (!_lastValues.ContainsKey(tag))
        {
            await _publishChannel.Writer.WriteAsync((tag, value, MqttQualityOfServiceLevel.AtLeastOnce, false));
            _lastValues.TryAdd(tag, value);
        }
    }
    // 优化后的批量发布策略
    public async Task OnDataChangedBulk(IEnumerable<(string tag, string value)> dataPairs)
    {
        var batch = new List<(string topic, string payload, MqttQualityOfServiceLevel qos, bool retain)>();
        
        foreach (var (tag, value) in dataPairs)
        {
            if (_lastValues.TryGetValue(tag, out var lastValue) && lastValue != value)
            {
                batch.Add((tag, value, MqttQualityOfServiceLevel.AtLeastOnce, false));
                _lastValues[tag] = value;
            }
            else if (!_lastValues.ContainsKey(tag))
            {
                batch.Add((tag, value, MqttQualityOfServiceLevel.AtLeastOnce, false));
                _lastValues.TryAdd(tag, value);
            }
        }

        if (batch.Any())
        {
            await _mqttClient.PublishBulkAsync(batch);
        }
    }
    // 阈值发布策略，当数据值达到或超过预设的阈值时进行发布
    public void OnThresholdReached(string tag, string value, double threshold, Comparison<double> comparison)
    {
        if (double.TryParse(value, out double numericValue)) {
            bool shouldPublish = comparison(numericValue, threshold) > 0;
            if (shouldPublish) {
                _mqttClient.PublishAsync(tag, value).Wait();
                _lastValues[tag] = value;
            }
        }
    }

    // 时间窗口发布策略，在指定的时间窗口内，只发布最新的数据
    public void OnTimeWindow(string tag, string value, TimeSpan window)
    {
        if (_lastPublishTimes.TryGetValue(tag, out var lastInfo)) {
            if (DateTime.UtcNow - lastInfo.lastPublishTime > window) {
                _mqttClient.PublishAsync(tag, value).Wait();
                _lastPublishTimes[tag] = (DateTime.UtcNow, value);
                _lastValues[tag] = value;
            }
            else {
                _lastPublishTimes[tag] = (lastInfo.lastPublishTime, value);
            }
        }
        else {
            _mqttClient.PublishAsync(tag, value).Wait();
            _lastPublishTimes[tag] = (DateTime.UtcNow, value);
            _lastValues.TryAdd(tag, value);
        }
    }

    // 累积发布策略，当数据累积到一定数量或达到一定条件时进行发布
    public async Task OnAccumulation(string tag, string value)
    {
        if (!_accumulatedData.TryGetValue(tag, out var dataList)) {
            dataList = new List<string>();
            _accumulatedData[tag] = dataList;
        }

        dataList.Add(value);

        if (dataList.Count >= AccumulationThreshold) {
            var combinedPayload = string.Join(",", dataList);
            await _mqttClient.PublishAsync(tag, combinedPayload);
            dataList.Clear();
            _lastValues[tag] = value;
        }
    }

    // 优先级发布策略，根据数据的优先级进行发布
    public async Task OnPriorityPublish(int minPriority)
    {
        var highPriorityData = _priorityData.Where(kv => kv.Value.priority >= minPriority).ToList();
        foreach (KeyValuePair<string, (string value, int priority)> item in highPriorityData) {
            string tag = item.Key;
            string value = item.Value.value;
            await _mqttClient.PublishAsync(tag, value);
            _lastValues[tag] = value;
        }
    }

    // 设置数据优先级
    public void SetPriority(string tag, string value, int priority)
    {
        _priorityData[tag] = (value, priority);
    }

    // 定时发布策略
    public void StartTimedPublish(TimeSpan interval, string topic, string payload, MqttQualityOfServiceLevel qos = MqttQualityOfServiceLevel.AtLeastOnce, bool retain = false)
    {
        _timer = new Timer(async _ => {
            await _mqttClient.PublishAsync(topic, payload, qos, retain);
        }, null, TimeSpan.Zero, interval);
    }
    // 并行定时批量发布策略
    public void StartTimedPublishBulkParallel(
        TimeSpan interval,
        IEnumerable<(string topic, string payload, MqttQualityOfServiceLevel qos, bool retain)> messages,
        int maxParallel = 8)
    {
        var messageArray = messages.ToArray();
        _timer = new Timer(async _ => 
        {
            try
            {
                int batchSize = 1;
                if (messageArray.Length > 100)
                    batchSize = messageArray.Length / 100;

                var ranges = Partitioner.Create(0, messageArray.Length, batchSize);
                Parallel.ForEach(ranges, new ParallelOptions { MaxDegreeOfParallelism = maxParallel }, 
                    range =>
                    {
                        // 转换为KeyValuePair格式
                        var batch = messageArray
                            .Skip(range.Item1)
                            .Take(range.Item2 - range.Item1)
                            .Select(x => new KeyValuePair<string, string>(x.topic, x.payload));
                        
                        _mqttClient.PublishBatchParallelAsync(batch).Wait();
                    });
            }
            catch (Exception ex)
            {
                Log.Error(ex, "并行定时批量发布失败");
            }
        }, null, TimeSpan.Zero, interval);
    }
    // 定时批量发布策略
    public void StartTimedPublishBulk(TimeSpan interval, IEnumerable<(string topic, string payload, MqttQualityOfServiceLevel qos, bool retain)> messages)
    {
        // 将消息列表转换为数组，避免多次枚举
        var messageArray = messages.ToArray();
        // 使用高性能定时器
        _timer = new Timer(async _ => 
        {
            try
            {
                // 分批次发布，
                int batchSize = 1;
                if (messageArray.Length < 100)
                     batchSize = messageArray.Length / 2;

                else if (messageArray.Length < 1000)
                    batchSize = messageArray.Length / 4;

                else if (messageArray.Length < 6000)
                    batchSize = messageArray.Length / 8;

                else if (messageArray.Length < 10000)
                    batchSize = messageArray.Length / 10;

                else if (messageArray.Length < 100000)
                    batchSize = messageArray.Length / 50;

                else if (messageArray.Length < 1000000)
                    batchSize = messageArray.Length / 500;

                for (int i = 0; i < messageArray.Length; i += batchSize)
                {
                    var batch = messageArray.Skip(i).Take(batchSize);
                    await _mqttClient.PublishBulkAsync(batch);
                }
            }
            catch (Exception ex)
            {
                Log.Error(ex, "定时批量发布失败");
            }
        }, null, TimeSpan.Zero, interval);
    }

    public void StopTimedPublish()
    {
        _timer?.Dispose();
    }
}
